package thread.pool.core.executor;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.*;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.CharsetUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.core.Ordered;
import org.springframework.util.CollectionUtils;
import thread.pool.core.biz.ExecutorBiz;
import thread.pool.core.biz.ExecutorBizImpl;
import thread.pool.core.util.GsonTool;
import thread.pool.core.util.HttpResult;
import thread.pool.core.util.ThrowableUtil;
import thread.pool.core.vo.UpdateVo;

import java.util.Map;
import java.util.concurrent.*;

/**
 * 内置通信服务持有类 将此类注入到容器中即可
 * 为确保单例bean 实现了接口SmartInitializingSingleton 在本地bean构建结束之后去去执行afterSingletonsInstantiated方法
 */
public class EmbedServerHolder implements ApplicationContextAware, SmartInitializingSingleton, DisposableBean, Ordered {

    // 默认端口9000
    private Integer port = 9000;
    // 内置服务
    private EmbedServerHolder.EmbedServer embedServer;
    // 所有的已经创建的动态线程池
    private static Map<String, DynamicThreadPoolExecutor> beanMap = new ConcurrentHashMap<>(4);
    private static final Map<String, DynamicThreadPoolExecutor> poolNameMap = new ConcurrentHashMap<>(4);
    private static ApplicationContext applicationContext;

    public EmbedServerHolder() {
    }


    /**
     * 单例bean实例化成功之后执行
     */
    public void afterSingletonsInstantiated() {

        // 从容器中找出程序员注入的动态下线程池
        this.obtainAllDynamicThreadPoolExecutor();

        // 初始化内置服务
        this.initEmbedServer();
    }

    /**
     * 获取所有的动态线程池
     */
    private void obtainAllDynamicThreadPoolExecutor() {
        beanMap = getBeansOfType(DynamicThreadPoolExecutor.class);
        if (CollectionUtils.isEmpty(beanMap)) {
            throw new RuntimeException("未定义动态线程池，服务未能正常启动 请检查线程池配置");
        }

        beanMap.forEach((k, v) -> {
            // 根据不同的bean名称作为key 放置新map
            poolNameMap.put(v.getPoolName(), v);
        });
    }

    public DynamicThreadPoolExecutor getHelperOfBeanName(String name) {
        return beanMap.get(name);
    }

    public DynamicThreadPoolExecutor getHelperOfPoolName(String name) {
        return poolNameMap.get(name);
    }

    private void initEmbedServer() {
        if (this.embedServer == null) {
            this.embedServer = new EmbedServer();
            if (this.port < 0) {
                this.port = 9000;
            }

            this.embedServer.start(this.port);
        }
    }

    public void closeServer() {
        this.embedServer.stop();
    }

    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        EmbedServerHolder.applicationContext = applicationContext;
    }

    public static ApplicationContext getApplicationContext() {
        return applicationContext;
    }

    public static <T> T getBean(String beanName, Class<T> tClass) {
        return applicationContext.getBean(beanName, tClass);
    }

    /**
     * 查找bean
     *
     * @param tClass
     * @param <T>
     * @return
     */
    public static <T> Map<String, T> getBeansOfType(Class<T> tClass) {
        return applicationContext.getBeansOfType(tClass);
    }

    public int getOrder() {
        return 2;
    }

    public Integer getPort() {
        return this.port;
    }

    public void setPort(Integer port) {
        this.port = port;
    }

    @Override
    public void destroy() throws Exception {
        closeServer();
    }

    /**
     * 内嵌服务
     */
    private static class EmbedServer {
        private static final Logger logger = LoggerFactory.getLogger(EmbedServer.class);

        private Thread thread;

        // 业务线程池
        private static ThreadPoolExecutor bizThreadPool;

        private void start(int port) {
            // 这里新起一个线程用去执行通信
            this.thread = new Thread(() -> {
                final ExecutorBizImpl executorBiz = new ExecutorBizImpl();
                bizThreadPool = new ThreadPoolExecutor(2, 10, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<>(100), new ThreadFactory() {
                    public Thread newThread(Runnable r) {
                        return new Thread(r, "thread-pool-rpc, EmbedServer bizThreadPool-" + r.hashCode());
                    }
                }, /* 拒绝策略*/
                        (r, executor) -> {
                            throw new RuntimeException("thread-pool-util, EmbedServer bizThreadPool is EXHAUSTED!");
                        });


                NioEventLoopGroup nioEventLoopGroup1 = new NioEventLoopGroup();
                NioEventLoopGroup nioEventLoopGroup2 = new NioEventLoopGroup();
                ServerBootstrap bootstrap = new ServerBootstrap();
                bootstrap.group(nioEventLoopGroup1, nioEventLoopGroup2).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer<SocketChannel>() {
                    public void initChannel(SocketChannel channel) {
                        channel.pipeline()
                                .addLast(new ChannelHandler[]{new IdleStateHandler(0L, 0L, 90L, TimeUnit.SECONDS)})
                                .addLast(new ChannelHandler[]{new HttpServerCodec()}).addLast(new ChannelHandler[]{new HttpObjectAggregator(5242880)})
                                .addLast(new EmbedHttpServerHandler(bizThreadPool, executorBiz));
                    }

                }).childOption(ChannelOption.SO_KEEPALIVE, Boolean.TRUE);
                try {
                    ChannelFuture future = bootstrap.bind(port).sync();
                    ChannelFuture channelFuture = future.channel().closeFuture();
                    channelFuture.sync();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
            this.thread.setDaemon(true);
            this.thread.start();
            logger.info(">>>>>>>>>>>>>>>>>内嵌服务启动成功，端口：{}", port);
        }

        /**
         * 停止包括了两个动作 1 监听http请求的线程 2 业务执行线程池
         */
        public void stop() {
            if (this.thread != null && this.thread.isAlive())
                this.thread.interrupt();
            bizThreadPool.shutdown();
            logger.info(">>>>>>>>>内置服务销毁成功");
        }

        /**
         * 请求处理器
         */
        static class EmbedHttpServerHandler extends SimpleChannelInboundHandler<FullHttpRequest> {
            private static final Logger logger = LoggerFactory.getLogger(EmbedHttpServerHandler.class);

            private final ThreadPoolExecutor bizThreadPool;

            private final ExecutorBiz executorBiz;

            public EmbedHttpServerHandler(ThreadPoolExecutor bizThreadPool, ExecutorBiz executorBiz) {
                this.bizThreadPool = bizThreadPool;
                this.executorBiz = executorBiz;
            }

            /**
             * 处理http请求
             *
             * @param ctx ctx
             */
            protected void channelRead0(final ChannelHandlerContext ctx, FullHttpRequest msg) throws Exception {
                final String requestData = msg.content().toString(CharsetUtil.UTF_8);
                final String uri = msg.uri();
                final HttpMethod httpMethod = msg.method();
                final boolean keepAlive = HttpUtil.isKeepAlive(msg);

                this.bizThreadPool.execute(() -> {
                    Object responseObj = this.process(httpMethod, uri, requestData);
                    String responseJson = GsonTool.toJson(responseObj);
                    this.writeResponse(ctx, keepAlive, responseJson);
                });
            }

            /**
             * 响应数据
             */
            private void writeResponse(ChannelHandlerContext ctx, boolean keepAlive, String responseJson) {
                DefaultFullHttpResponse defaultFullHttpResponse = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK, Unpooled.copiedBuffer(responseJson, CharsetUtil.UTF_8));
                defaultFullHttpResponse.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/html;charset=UTF-8");
                defaultFullHttpResponse.headers().set(HttpHeaderNames.CONTENT_LENGTH, defaultFullHttpResponse.content().readableBytes());
                if (keepAlive)
                    defaultFullHttpResponse.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
                ctx.writeAndFlush(defaultFullHttpResponse);
            }

            /**
             * 处理请求业务
             */
            private Object process(HttpMethod httpMethod, String uri, String requestData) {
                if (HttpMethod.POST != httpMethod)
                    return new HttpResult(HttpResult.FAIL_CODE, "请求方式不支持，暂只支持POST请求");
                if (uri == null || uri.trim().length() == 0)
                    return new HttpResult(HttpResult.FAIL_CODE, "请求不合法 url为空");
                try {
                    UpdateVo updateVo;
                    switch (uri) {
                        case "/check":
                            return this.executorBiz.check();
                        case "/update":
                            updateVo = GsonTool.fromJson(requestData, UpdateVo.class);
                            return this.executorBiz.update(updateVo);
                    }
                } catch (Exception e) {
                    logger.error(e.getMessage(), e);
                    return new HttpResult(HttpResult.FAIL_CODE, "request error:" + ThrowableUtil.toString(e));
                }
                return new HttpResult();
            }
        }
    }


}

